package com.lg.monitor

import java.sql.{Connection, DriverManager, PreparedStatement, Statement}

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter

object MysqlWriter {
  //获取连接

  val username = "hive"
  val password = "12345678"
  val url = "jdbc:mysql://zb26107:3306/ebiz?useUnicode=true&characterEncoding=utf-8&useSSL=false"

  def getConnection:Connection={
    DriverManager.getConnection(url, username, password)
  }

}

class MysqlWriter extends ForeachWriter[BusInfo]{
  var conn: Connection = _

  override def open(partitionId: Long, epochId: Long): Boolean = {

    conn = MysqlWriter.getConnection

    true
  }

  override def process(value: BusInfo): Unit = {
    val sql = "insert into lg_bus_lglat values (?, ?)"
    val ps: PreparedStatement = conn.prepareStatement(sql)

    //把数据写入mysql ,k:v形式  k是deployNum，v是lglat
    val lglat: String = value.lglat
    val deployNum = value.deployNum

    ps.setString(1, deployNum)
    ps.setString(2, lglat)
    ps.executeUpdate()


  }

  override def close(errorOrNull: Throwable): Unit = {
    conn.close();
  }
}
